package io.kiki.sba.registry.server.metadata.remoting.meta;

import io.kiki.sba.registry.api.ChannelHandler;
import io.kiki.sba.registry.api.exchange.DataExchanger;
import io.kiki.sba.registry.api.exchange.RequestException;
import io.kiki.sba.registry.api.exchange.message.Request;
import io.kiki.sba.registry.api.exchange.message.Response;
import io.kiki.sba.registry.api.exchange.message.Response.ResultStatus;
import io.kiki.sba.registry.common.model.store.Url;
import io.kiki.sba.registry.server.metadata.MetadataLeaderService;
import io.kiki.sba.registry.server.metadata.bootstrap.config.ServerConfig;
import io.kiki.sba.registry.server.shared.remoting.ClientSideExchanger;
import lombok.Getter;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.Collection;
import java.util.Collections;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;


public class MetaNodeExchange extends ClientSideExchanger {

    private static final Logger logger = LoggerFactory.getLogger(MetaNodeExchange.class);
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    @Getter
    protected volatile String metaLeader;
    @Autowired
    private ServerConfig serverConfig;
    @Autowired
    private MetadataLeaderService metadataLeaderService;

    public MetaNodeExchange() {
        super(DataExchanger.META_SERVER_TYPE);
    }

    @Override
    public int getRpcTimeoutMillis() {
        return serverConfig.getMetaNodeExchangeTimeoutMillis();
    }

    @Override
    public int getServerPort() {
        return serverConfig.getMetaServerPort();
    }

    @Override
    protected Collection<ChannelHandler> getClientHandlers() {
        return Collections.emptyList();
    }

    @Override
    public int getConnNum() {
        return 3;
    }

    public Response sendRequest(Object requestBody) throws RequestException {
        final String newLeader = metadataLeaderService.getLeader();
        if (StringUtils.isBlank(newLeader)) {
            logger.error("[sendRequest] meta leader is empty.");
            return () -> ResultStatus.FAILED;
        }

        if (!StringUtils.equals(metaLeader, newLeader) || nettyDataExchanger.getClient(serverType) == null) {
            setLeaderAndConnect(newLeader);
        }

        Request request = new Request() {
            @Override
            public Object getRequestBody() {
                return requestBody;
            }

            @Override
            public Url getRequestUrl() {
                return new Url(newLeader, getServerPort());
            }
        };
        return request(request);
    }

    private void setLeaderAndConnect(String newLeader) {
        String removed = metaLeader;
        try {
            lock.writeLock().lock();
            metaLeader = newLeader;
        } finally {
            lock.writeLock().unlock();
        }

        try {
            logger.info("[setLeaderAndConnect][reset leader when heartbeat] connect meta leader: {}", newLeader, removed);
            connect(new Url(newLeader, serverConfig.getMetaServerPort()));

        } catch (Throwable th) {
            logger.error("[setLeaderAndConnect]", th);
        }
    }


}
